package com.tuandai.log.logback.appender;

import ch.qos.logback.core.OutputStreamAppender;
import ch.qos.logback.core.encoder.LayoutWrappingEncoder;
import ch.qos.logback.core.status.ErrorStatus;
import com.tuandai.log.kafka.config.Slf4jKafkaConfig;
import com.tuandai.log.kafka.config.Slf4jKafkaConfigSimpleFactory;
import com.tuandai.log.kafka.producer.KafkaProducerProcessor;

/**
 *
 * Created by zhangzhenbin on 17-5-24.
 */
public class KafkaAppender<E> extends OutputStreamAppender<E> {
    protected static Slf4jKafkaConfig slf4jKafkaConfig;

    public KafkaAppender() throws Exception {
        slf4jKafkaConfig = Slf4jKafkaConfigSimpleFactory.getInstance().parse();
    }
    @Override
    public void start() {
        int errors = 0;
        if(slf4jKafkaConfig == null){
            addStatus(new ErrorStatus("Kafka Config is Invalid!", this));
            errors++;
        }
        // only error free appenders should be activated
        if (errors == 0) {
            started = true;
            KafkaProducerProcessor.init(slf4jKafkaConfig);
        }
    }
    /**
     * This method differentiates RollingFileAppender from its super class.
     */
    @Override
    protected void subAppend(E event) {
        LayoutWrappingEncoder<E> encoder = (LayoutWrappingEncoder<E>) this.getEncoder();
        String msg = encoder.getLayout().doLayout(event);
        KafkaProducerProcessor.send(slf4jKafkaConfig,msg);
    }
}
